package com.atguigu.api2

import com.atguigu.api.SensorReading
import org.apache.flink.streaming.api.functions.ProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

/**
 * @description: 高温流放在主流,低温流作为侧输出流
 * @time: 2020/7/9 17:38
 * @author: baojinlong
 **/
object SideOutputTest {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    // 读取文本流
    val inputStream: DataStream[String] = env.socketTextStream("123.56.230.239", 7777)
    val dataStream: DataStream[SensorReading] = inputStream.map(data => {
      val dataArray: Array[String] = data.split(",")
      SensorReading(dataArray(0), dataArray(1).toLong, dataArray(2).toDouble)
    })


    // 用ProcessFunction的侧输出流来实现分流操作
    val highTempStream: DataStream[SensorReading] = dataStream.process(new SplitTempProcessor(30.0))

    val lowTempStream: DataStream[(String, Double, Long)] = highTempStream.getSideOutput(new OutputTag[(String, Double, Long)]("low-temp"))


    // 打印
    highTempStream.print("high")
    lowTempStream.print("low")
    // 执行
    env.execute("side output job")

  }
}

/**
 * 数据一条条进来了
 *
 * @param threshold
 */
class SplitTempProcessor(threshold: Double) extends ProcessFunction[SensorReading, SensorReading] {
  override def processElement(value: SensorReading, context: ProcessFunction[SensorReading, SensorReading]#Context, collector: Collector[SensorReading]): Unit = {
    // 处理每一个元素,判断当前数据的阈值
    if (value.temperature > threshold) {
      collector.collect(value)
    } else {
      // 侧输出流
      context.output(new OutputTag[(String, Double, Long)]("low-temp"), (value.id, value.temperature, value.timestamp))
    }
  }
}